package com.example.flow

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import org.junit.Test


/**
 * 流的上下文测试
 * 1.上下文保存：Flow的收集总是在调用协程的上下文中发生，留的该属性成为上下文保存
 * 2.flow构建器中的代码遵循上下文保存属性，并且不允许从其他上下文中发射元素
 * 3.flowOn操作符，该函数用于更改流发射的上下文
 * 4.Flow可以通过launchIn指定收集执行的上下文，并且返回一个Job
 *
 * @author zhouronghua
 * @time 2022/1/5 10:17 下午
 */
class FlowContextTest {

    /**
     * 测试流的上下文保存
     *
     * @author zhouronghua
     * @time 2022/1/5 10:23 下午
     */
    @Test
    fun test_flow_context_save() = runBlocking {
        val scope = CoroutineScope(
            Job() + Dispatchers.IO + CoroutineName("我的协程")
        )
        var myFlow: Flow<Int>? = null
        scope.launch {
            try {
                myFlow = flow<Int> {
                    (1..5).forEach {
                        println("发射元素 $it ${Thread.currentThread().name}")
                        emit(it)
                    }
                }
            } catch (e: Exception) {
                e.printStackTrace()
            } finally {
                println("流发射结束")
            }

        }
        // 在其他协程收集，不会处理(采用scope.launch能够收集)
        val job = GlobalScope.launch {
            myFlow?.collect {
                println("Collect $it on ${Thread.currentThread().name}")
            }
        }
        job.join()
    }

    /**
     * 测试流的上下文保存
     * 在同一协程上下文能够接收流
     *
     * @author zhouronghua
     * @time 2022/1/5 10:23 下午
     */
    @Test
    fun test_flow_context_save_context() = runBlocking {
        val scope = CoroutineScope(
            Job() + Dispatchers.IO + CoroutineName("我的协程")
        )
        var myFlow: Flow<Int>? = null
        scope.launch {
            try {
                myFlow = flow<Int> {
                    (1..5).forEach {
                        println("发射元素 $it ${Thread.currentThread().name}")
                        emit(it)
                    }
                }
            } catch (e: Exception) {
                e.printStackTrace()
            } finally {
                println("流发射结束")
            }

        }
        println("在发射流的上下文收集")
        // 在其他协程收集，不会处理(采用scope.launch能够收集)
        val job = scope.launch {
            myFlow?.collect {
                println("Collect $it on ${Thread.currentThread().name}")
            }
        }
        job.join()
    }

    /**
     * 测试通过flowOn切换发射流的上下文
     * 说明：通过IO协程调度器发射，主线程接收流
     *
     * @author zhouronghua
     * @time 2022/1/5 10:37 下午
     */
    @Test
    fun test_flow_context_switch_with_flowOn() = runBlocking {
        val simpleFlow = flow<Int> {
            println("流开始发射")
            for (i in 1..5) {
                println("发射元素 $i ${Thread.currentThread().name}")
                delay(100)
                emit(i)
            }
        }
        simpleFlow.flowOn(Dispatchers.IO)
            .collect {
                delay(200)
                println("接收元素 $it ${Thread.currentThread().name}")
            }
    }

    /**
     * 构建一个流每个100ms发射，通过默认协程调度器
     *
     * @author zhouronghua
     * @time 2022/1/5 10:49 下午
     */
    fun events() = (1..3).asFlow()
        .onEach {
            delay(100)
            println("发射元素$it ${Thread.currentThread().name}")
        }.flowOn(Dispatchers.Default)

    /**
     * 在指定协程收集流
     * 流可以通过launchIn指定收集流的协程上下文
     *
     * @author zhouronghua
     * @time 2022/1/5 10:52 下午
     */
    @Test
    fun test_flow_collect_launch() = runBlocking {
        // 在主协程收集流
        println("指定在主协程收集流")
        events().onEach {
            println("Event: $it ")
        }.collect {
            println("收集到元素 $it ${Thread.currentThread().name}")
        }
        println("\n指定在IO协程收集流元素")
        // 指定IO协程协程收集流
        val job = events().onEach {
            println("Event: $it ${Thread.currentThread().name}")
        }.launchIn(CoroutineScope(Dispatchers.IO))
        job.join()
    }
}